package com.binlog.util;

import cn.hutool.core.util.StrUtil;
import com.binlog.config.BinLogDbProperty;
import com.binlog.config.BinLogDbPropertys;
import com.binlog.module.DatebaseInfo;
import com.binlog.module.TableColum;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
public class MysqlUtils {

    /**
     * 获取columns集合
     */
    public static Map<String, TableColum> getColMap(BinLogDbProperty conf, String table) {
        Connection connection = DruidDataSource.getConn(conf.getDataSourceName());
        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            // 执行sql
            String preSql = "SELECT TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME, DATA_TYPE, ORDINAL_POSITION, COLUMN_KEY FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = ? and TABLE_NAME = ?";
            ps = connection.prepareStatement(preSql);
            ps.setString(1, conf.getSchema());
            ps.setString(2, table);
            rs = ps.executeQuery();
            Map<String, TableColum> map = new HashMap<>(rs.getRow());
            while (rs.next()) {
                String schema = rs.getString("TABLE_SCHEMA");
                String tableName = rs.getString("TABLE_NAME");
                String column = rs.getString("COLUMN_NAME");
                String index = rs.getString("COLUMN_KEY");
                int idx = rs.getInt("ORDINAL_POSITION");
                String dataType = rs.getString("DATA_TYPE");
                if (column != null && idx >= 1) {
                    // sql的位置从1开始
                    map.put(column, new TableColum(schema, tableName, idx - 1, column, dataType, index));
                }
            }
            // 使用Stream API进行排序和转换为有序的LinkedHashMap
            return map.entrySet().stream()
                    .sorted(Comparator.comparing(entry -> entry.getValue().getInx()))
                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, (oldValue, newValue) -> oldValue, LinkedHashMap::new));
        } catch (Exception e) {
            log.error("load db conf error, db_table={}:{} ", conf.getSchema(), table, e);
        } finally {
            DruidDataSource.close(connection, ps, rs);
        }
        return null;
    }

    /**
     * 创建数据库sql
     * @param conf
     * @return
     */
    public static String getCreateSchemaSql(BinLogDbProperty conf) {
        Connection connection = DruidDataSource.getConn(conf.getDataSourceName());
        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            // 执行sql
            String preSql = "SHOW CREATE DATABASE " + conf.getSchema();
            ps = connection.prepareStatement(preSql);
            rs = ps.executeQuery();
            String sql = null;
            while (rs.next()) {
                // 从结果集中读取数据
                sql = rs.getString("Create Database");
            }
            return sql;
        } catch (Exception e) {
            log.error("getCreateSchemaSql failed: {}", conf.getSchema());
        } finally {
            DruidDataSource.close(connection, ps, rs);
        }
        return null;
    }

    /**
     * 创建数据库sql
     * @param conf
     * @param table
     * @return
     */
    private static String getCreateTableSql(BinLogDbProperty conf, String table) {
        Connection connection = DruidDataSource.getConn(conf.getDataSourceName());
        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            // 执行sql
            String preSql = "SHOW CREATE TABLE " + table;
            ps = connection.prepareStatement(preSql);
            rs = ps.executeQuery();
            String sql = null;
            while (rs.next()) {
                // 从结果集中读取数据
                sql = rs.getString("Create Table");
            }
            return sql;
        } catch (Exception e) {
            log.error("getCreateSchemaSql failed: {}", conf.getSchema());
        } finally {
            DruidDataSource.close(connection, ps, rs);
        }
        return null;
    }

    /**
     * 判断数据库是否存在
     * @param conf
     * @param schema
     * @return
     */
    private static Boolean isExitsSchema(BinLogDbProperty conf, String schema) {
        Connection connection = DruidDataSource.getConn(conf.getDataSourceName());
        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            // 执行sql
            String preSql = "SHOW DATABASES LIKE '" + schema + "'";
            ps = connection.prepareStatement(preSql);
            rs = ps.executeQuery();
            boolean exits = false;
            while (rs.next()) {
                // 从结果集中读取数据
                exits = StringUtils.isNotEmpty(rs.getString(1));
            }
            return exits;
        } catch (Exception e) {
            log.error("isExitsSchema failed: {} e:{}", conf.getSchema(), e.getMessage(), e);
        } finally {
            DruidDataSource.close(connection, ps, rs);
        }
        return false;
    }


    /**
     * 判断数据库表是否存在
     * @param conf
     * @param schema
     * @return
     */
    private static Boolean isExitsTable(BinLogDbProperty conf, String table, String schema) {
        Connection connection = DruidDataSource.getConn(conf.getDataSourceName());
        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            // 执行sql
            String preSql = "select * from information_schema.`TABLES` where table_schema = ? and table_Name = ?";
            ps = connection.prepareStatement(preSql);
            ps.setString(1, schema);
            ps.setString(2, table);
            rs = ps.executeQuery();
            boolean exits = false;
            while (rs.next()) {
                // 从结果集中读取数据
                exits = StringUtils.isNotEmpty(rs.getString(1));
            }
            return exits;
        } catch (Exception e) {
            log.error("isExitsTable failed: {} e:{}", conf.getSchema());
        } finally {
            DruidDataSource.close(connection, ps, rs);
        }
        return false;
    }

    /**
     * 将逗号拼接字符串转List 并去除首尾空格
     *
     * @param tables
     * @return
     */
    public static List<String> getTableByStr(String tables) {
        if (StrUtil.isEmpty(tables)) {
            return Lists.newArrayList();
        }

        return Arrays.asList(tables.split(",")).stream().map(String::trim).collect(Collectors.toList());
    }

    /**
     * 获取备份库信息
     *
     * @param binLogDbProperty
     * @param binLogDbProperty
     * @return
     */
    public static DatebaseInfo getDatebaseInfo(BinLogDbProperty binLogDbProperty, BinLogDbProperty backupsDatasource) {
        DatebaseInfo datebaseInfo = new DatebaseInfo();
        datebaseInfo.setCreateDatebaseSql(MysqlUtils.getCreateSchemaSql(binLogDbProperty));
        datebaseInfo.setExitsDatebase(MysqlUtils.isExitsSchema(backupsDatasource, binLogDbProperty.getSchema()));

        datebaseInfo.setCreateTableSql(
                MysqlUtils.getTableByStr(binLogDbProperty.getTable()).stream().collect(Collectors.toMap(
                        table -> table,
                        table -> MysqlUtils.getCreateTableSql(binLogDbProperty, table)
                ))
        );

        datebaseInfo.setExitsTable(
                MysqlUtils.getTableByStr(binLogDbProperty.getTable()).stream().collect(Collectors.toMap(
                        table -> table,
                        table -> MysqlUtils.isExitsTable(backupsDatasource, table, binLogDbProperty.getSchema())
                ))
        );

        return datebaseInfo;
    }


    /**
     * 直接执行sql
     * @param sql
     */
    public static void executBackSql(String sql, String schema) {
        BinLogDbProperty backupsDatasource = SpringContextUtils.getBean(BinLogDbPropertys.class).getBackupsDatasource();
        if (null == backupsDatasource) {
            return;
        }
        if (StringUtils.isEmpty(schema)) {
            DruidDataSource.execute(backupsDatasource.getDataSourceName(), sql);
        } else {
            DruidDataSource.execute(backupsDatasource.getDataSourceName() + "_" + schema, sql);
        }

    }
}
